package com.ffcs.crmd.tsp.server.job.service;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ffcs.crmd.tsp.api.dto.job.TspJobExecutionContext;
import com.ffcs.crmd.tsp.common.constant.JobRefireLevel;
import com.ffcs.crmd.tsp.common.constant.LoggerName;
import com.ffcs.crmd.tsp.server.job.info.JobDetailShell;
import com.ffcs.crmd.tsp.server.job.info.JobDetailShellChannelPairInfo;
import com.ffcs.crmd.tsp.server.job.info.JobDetailTrack;

import io.netty.channel.Channel;

/**
 * 
 * 功能说明:任务处理服务
 *
 * @author ZHONGFUHUA
 * 
 * @since 1.0.0-SNAPSHOT
 *
 */
public class JobProcessQueue {
    
    /**
     * 日志
     */
    private final static Logger                               LOG                   = LoggerFactory.getLogger(LoggerName.ServerLoggerName);
    
    /**
     * 子任务的运行次数
     */
    private volatile Map<String, AtomicLong>                  runtimesMap           = new HashMap<String, AtomicLong>();
    
    /**
     * quartz已触发的子任务数量
     */
    private AtomicLong                                        currentSize           = new AtomicLong();
    
    /** 
     * 任务名
     */
    private final String                                      jobName;
    
    /**
     * 通道失效超时时间
     */
    private long                                              timeout               = 5 * 60 * 1000;
    
    /**
     * 超时策略
     */
    private short                                             timeoutPolicy         = 0;
    
    /**
     * 任务数量缓冲区大小
     */
    private int                                               bufferLimitSize       = 5000;
    
    /**
     * 死任务重置延迟时间 -1:不延迟
     */
    private long                                              resetDeathDelayTime   = -1;
    
    /**
     * 子任务操作锁
     */
    private ReadWriteLock                                     lockJobDetail         = new ReentrantReadWriteLock();
    
    /**
     * 重试子任务操作锁
     */
    private ReadWriteLock                                     lockJobDetailRetry    = new ReentrantReadWriteLock();
    
    /**
     * 死子任务操作锁
     */
    private ReadWriteLock                                     lockJobDetailDeath    = new ReentrantReadWriteLock();
    
    /**
     * 正在执行子任务操作锁
     */
    private ReadWriteLock                                     lockJobDetailRunning  = new ReentrantReadWriteLock(true);
    
    /**
     * 任务轨迹操作锁
     */
    private ReadWriteLock                                     lockJobDetailTrack    = new ReentrantReadWriteLock(true);
    
    /**
     * 待运行任务
     */
    private volatile LinkedList<JobDetailShell>               jobDetailQueue        = new LinkedList<JobDetailShell>();
    
    /**
     * 待重试运行任务
     */
    private volatile LinkedList<JobDetailShell>               jobDetailRetryQueue   = new LinkedList<JobDetailShell>();
    
    /**
     * 死子任务<子任务名, 死子任务>
     */
    private volatile Map<String, LinkedList<JobDetailShell>>  jobDetailDeathTable   = new HashMap<String, LinkedList<JobDetailShell>>();
    
    /**
     * 子任务转移轨迹表
     */
    private volatile Map<String, JobDetailTrack>              jobDetailTrackTable   = new HashMap<String, JobDetailTrack>();
    
    /**
     * 正在执行任务
     */
    private volatile Map<Channel, LinkedList<JobDetailShell>> jobDetailRunningTable = new HashMap<Channel, LinkedList<JobDetailShell>>();
    
    public JobProcessQueue(String jobName) {
        this.jobName = jobName;
    }
    
    /**
     * 
     * 功能说明:添加待执行的子任务
     * 
     * @param jobDetailRunInfo 添加待执行的子任务
     */
    public void putJobDetail(JobDetailShell jobDetailRunInfo) {
        lockJobDetail.writeLock().lock();
        try {
            if (currentSize.get() <= bufferLimitSize) {
                jobDetailQueue.add(jobDetailRunInfo);
                currentSize.incrementAndGet();
            } else {
                LOG.warn("JOB-PROCESS-QUEUE: buffer[{} > {}] is over will flow control.", currentSize, bufferLimitSize);
            }
        } finally {
            lockJobDetail.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:添加重试运行的子任务
     * 
     * @param jobDetailShell 添加待执行的子任务
     */
    public void putJobDetailRetry(JobDetailShell jobDetailShell) {
        lockJobDetailRetry.writeLock().lock();
        try {
            jobDetailShell.updateFireTime();
            jobDetailRetryQueue.add(jobDetailShell);
        } finally {
            lockJobDetailRetry.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:已发送执行端处理的子任务放置临时缓存
     * 
     * @param channel 通道
     * @param jobDetailShell 子任务内核
     */
    public void putJobDetailRunning(Channel channel, JobDetailShell jobDetailShell) {
        lockJobDetailTrack.writeLock().lock();
        try {
            JobDetailTrack jobDetailTrack = jobDetailTrackTable.get(jobDetailShell.getId());
            if (jobDetailTrack == null) {
                jobDetailTrack = new JobDetailTrack();
            }
            jobDetailTrack.putChannel(channel, jobDetailShell);
            jobDetailTrackTable.put(jobDetailShell.getId(), jobDetailTrack);
        } finally {
            lockJobDetailTrack.writeLock().unlock();
        }
        
        lockJobDetailRunning.writeLock().lock();
        try {
            LinkedList<JobDetailShell> jobDetailShellRunningList = jobDetailRunningTable.get(channel);
            if (jobDetailShellRunningList == null) {
                jobDetailShellRunningList = new LinkedList<JobDetailShell>();
            }
            jobDetailShellRunningList.add(jobDetailShell);
            jobDetailRunningTable.put(channel, jobDetailShellRunningList);
        } finally {
            lockJobDetailRunning.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:添加死子任务
     * 
     * @param jobDetailShell
     */
    public void putJobDetailDeath(JobDetailShell jobDetailShell) {
        lockJobDetailDeath.writeLock().lock();
        try {
            String jobDetailName = jobDetailShell.getContext().getTspJobInstance().getName();
            LinkedList<JobDetailShell> jobDetailShellDeathQueue = jobDetailDeathTable.get(jobDetailName);
            if (jobDetailShellDeathQueue == null) {
                jobDetailShellDeathQueue = new LinkedList<JobDetailShell>();
            }
            jobDetailShellDeathQueue.add(jobDetailShell);
            jobDetailDeathTable.put(jobDetailName, jobDetailShellDeathQueue);
        } finally {
            lockJobDetailDeath.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:
     *
     */
    public void deleteJob() {
        lockJobDetail.writeLock().lock();
        lockJobDetailDeath.writeLock().lock();
        lockJobDetailRetry.writeLock().lock();
        lockJobDetailRunning.writeLock().lock();
        lockJobDetailTrack.writeLock().lock();
        try {
            Iterator<JobDetailShell> jobDetailIt = jobDetailQueue.iterator();
            while (jobDetailIt.hasNext()) {
                JobDetailShell jobDetailShell = jobDetailIt.next();
                jobDetailIt.remove();
                jobDetailShell.callBack();
            }
            
            Iterator<JobDetailShell> jobDetailRetryIt = jobDetailRetryQueue.iterator();
            while (jobDetailRetryIt.hasNext()) {
                JobDetailShell jobDetailShell = jobDetailRetryIt.next();
                jobDetailRetryIt.remove();
                jobDetailShell.callBack();
            }
            
            Iterator<LinkedList<JobDetailShell>> jobDetailRunningIt = jobDetailRunningTable.values().iterator();
            while (jobDetailRunningIt.hasNext()) {
                LinkedList<JobDetailShell> jobDetailRunningList = jobDetailRunningIt.next();
                Iterator<JobDetailShell> jobDetailRunningListIt = jobDetailRunningList.iterator();
                while (jobDetailRunningListIt.hasNext()) {
                    JobDetailShell jobDetailShell = jobDetailRunningListIt.next();
                    jobDetailRunningListIt.remove();
                    removeJobDetailTrack(jobDetailShell);
                    jobDetailShell.callBack();
                }
            }
            
            Iterator<LinkedList<JobDetailShell>> jobDetailDeathIt = jobDetailDeathTable.values().iterator();
            while (jobDetailDeathIt.hasNext()) {
                LinkedList<JobDetailShell> jobDetailDeathList = jobDetailDeathIt.next();
                Iterator<JobDetailShell> jobDetailDeathListIt = jobDetailDeathList.iterator();
                while (jobDetailDeathListIt.hasNext()) {
                    JobDetailShell jobDetailShell = jobDetailDeathListIt.next();
                    jobDetailDeathListIt.remove();
                    removeJobDetailTrack(jobDetailShell);
                    jobDetailShell.callBack();
                }
            }
            
            jobDetailQueue.clear();
            jobDetailRetryQueue.clear();
            jobDetailRunningTable.clear();
            jobDetailDeathTable.clear();
            jobDetailTrackTable.clear();
        } finally {
            lockJobDetailTrack.writeLock().unlock();
            lockJobDetailRunning.writeLock().unlock();
            lockJobDetailRetry.writeLock().unlock();
            lockJobDetailDeath.writeLock().unlock();
            lockJobDetail.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:
     *
     */
    public void deleteJobDetail(String jobDetailName) {
        lockJobDetail.writeLock().lock();
        lockJobDetailDeath.writeLock().lock();
        lockJobDetailRetry.writeLock().lock();
        lockJobDetailRunning.writeLock().lock();
        lockJobDetailTrack.writeLock().lock();
        try {
            Iterator<JobDetailShell> jobDetailIt = jobDetailQueue.iterator();
            while (jobDetailIt.hasNext()) {
                JobDetailShell jobDetailShell = jobDetailIt.next();
                if (jobDetailShell.getJobDetailName().equals(jobDetailName)) {
                    jobDetailIt.remove();
                    jobDetailShell.callBack();
                }
            }
            
            Iterator<JobDetailShell> jobDetailRetryIt = jobDetailRetryQueue.iterator();
            while (jobDetailRetryIt.hasNext()) {
                JobDetailShell jobDetailShell = jobDetailRetryIt.next();
                if (jobDetailShell.getJobDetailName().equals(jobDetailName)) {
                    jobDetailRetryIt.remove();
                    jobDetailShell.callBack();
                }
            }
            
            Iterator<LinkedList<JobDetailShell>> jobDetailRunningIt = jobDetailRunningTable.values().iterator();
            while (jobDetailRunningIt.hasNext()) {
                LinkedList<JobDetailShell> jobDetailRunningList = jobDetailRunningIt.next();
                Iterator<JobDetailShell> jobDetailRunningListIt = jobDetailRunningList.iterator();
                while (jobDetailRunningListIt.hasNext()) {
                    JobDetailShell jobDetailShell = jobDetailRunningListIt.next();
                    if (jobDetailShell.getJobDetailName().equals(jobDetailName)) {
                        jobDetailRunningListIt.remove();
                        removeJobDetailTrack(jobDetailShell);
                        jobDetailShell.callBack();
                    }
                }
            }
            
            Iterator<LinkedList<JobDetailShell>> jobDetailDeathIt = jobDetailDeathTable.values().iterator();
            while (jobDetailDeathIt.hasNext()) {
                LinkedList<JobDetailShell> jobDetailDeathList = jobDetailDeathIt.next();
                Iterator<JobDetailShell> jobDetailDeathListIt = jobDetailDeathList.iterator();
                while (jobDetailDeathListIt.hasNext()) {
                    JobDetailShell jobDetailShell = jobDetailDeathListIt.next();
                    if (jobDetailShell.getJobDetailName().equals(jobDetailName)) {
                        jobDetailDeathListIt.remove();
                        removeJobDetailTrack(jobDetailShell);
                        jobDetailShell.callBack();
                    }
                }
            }
        } finally {
            lockJobDetailTrack.writeLock().unlock();
            lockJobDetailRunning.writeLock().unlock();
            lockJobDetailRetry.writeLock().unlock();
            lockJobDetailDeath.writeLock().unlock();
            lockJobDetail.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:重试任务全部转移到待处理列表
     *
     */
    public void flushJobDetailRetry() {
        lockJobDetail.writeLock().lock();
        lockJobDetailRetry.writeLock().lock();
        try {
            List<JobDetailShell> removeList = new ArrayList<JobDetailShell>();
            for (JobDetailShell jobDetailShell : jobDetailRetryQueue) {
                if (jobDetailShell.getContext().getRefireCount() > 0) {
                    if (System.currentTimeMillis() - jobDetailShell.getFireTime().getTime() > JobRefireLevel
                        .getTime(jobDetailShell.getContext().getRefireCount())) {
                        removeList.add(jobDetailShell);
                    }
                } else {
                    removeList.add(jobDetailShell);
                }
            }
            jobDetailQueue.addAll(removeList);
            jobDetailRetryQueue.removeAll(removeList);
        } finally {
            lockJobDetailRetry.writeLock().unlock();
            lockJobDetail.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:标记该通道的运行子任务为离线状态
     * 
     * @param channel 通道
     */
    public void offlineJobDetailRunning(Channel channel) {
        lockJobDetailRunning.writeLock().lock();
        try {
            LinkedList<JobDetailShell> jobDetailShellList = jobDetailRunningTable.get(channel);
            if (CollectionUtils.isNotEmpty(jobDetailShellList)) {
                for (JobDetailShell jobDetailShell : jobDetailShellList) {
                    jobDetailShell.updateOffline();
                    LOG.warn("JOB-PROCESS-QUEUE: update job detail running [{}.{}] id=[{}] opaque=[{}] to offline.", jobName,
                        jobDetailShell.getJobDetailName(), jobDetailShell.getId(), jobDetailShell.getRemotingOpaque());
                }
            }
        } finally {
            lockJobDetailRunning.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param channel
     * @return
     */
    public List<JobDetailShell> getJobDetailOffline(Channel channel) {
        List<JobDetailShell> rjobDetailShellList = new ArrayList<JobDetailShell>();
        lockJobDetailRunning.writeLock().lock();
        try {
            LinkedList<JobDetailShell> jobDetailShellList = jobDetailRunningTable.get(channel);
            if (CollectionUtils.isNotEmpty(jobDetailShellList)) {
                rjobDetailShellList.addAll(jobDetailShellList);
            }
        } finally {
            lockJobDetailRunning.writeLock().unlock();
        }
        return rjobDetailShellList;
    }
    
    /**
     * 
     * 功能说明:标记该通道的运行子任务为关闭状态
     * 
     * @param channel 通道
     */
    public void shutdownJobDetailRunning(Channel channel) {
        lockJobDetailRunning.writeLock().lock();
        lockJobDetailRetry.writeLock().lock();
        try {
            LinkedList<JobDetailShell> jobDetailShellList = jobDetailRunningTable.remove(channel);
            if (CollectionUtils.isNotEmpty(jobDetailShellList)) {
                for (JobDetailShell jobDetailShell : jobDetailShellList) {
                    jobDetailShell.updateFireTime();
                    jobDetailRetryQueue.add(jobDetailShell);
                    LOG.warn("JOB-PROCESS-QUEUE: shutdown job detail running [{}.{}] id=[{}] and put retry to queue", jobName,
                        jobDetailShell.getJobDetailName(), jobDetailShell.getId());
                }
            }
        } finally {
            lockJobDetailRetry.writeLock().unlock();
            lockJobDetailRunning.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:删除死任务
     * 
     * @param jobDetailName 子任务名称
     * @param jobDetailShell 子任务运行内核
     */
    public boolean removeJobDetailDeath(JobDetailShell jobDetailShell) {
        boolean result = false;
        lockJobDetailDeath.writeLock().lock();
        try {
            List<JobDetailShell> jobDetailShellDeathQueue = jobDetailDeathTable.get(jobDetailShell.getContext().getTspJobInstance().getName());
            if (CollectionUtils.isNotEmpty(jobDetailShellDeathQueue)) {
                result = jobDetailShellDeathQueue.remove(jobDetailShell);
                if (result) {
                    LOG.info("JOB-PROCESS-QUEUE: release and remove a death job detail[{}] id=[{}] opaque=[{}]",
                        jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(), jobDetailShell.getRemotingOpaque());
                } else {
                    LOG.warn("JOB-PROCESS-QUEUE: release and remove a death job detail[{}] id=[{}] opaque=[{}] fail because isn't found.",
                        jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(), jobDetailShell.getRemotingOpaque());
                }
            }
        } finally {
            lockJobDetailDeath.writeLock().unlock();
        }
        return result;
    }
    
    /**
     * 
     * 功能说明:删除死任务
     * 
     * @param jobDetailName 子任务名称
     * @param jobDetailShell 子任务运行内核
     */
    private JobDetailShell removeJobDetailDeath(String jobDetailName, String id) {
        lockJobDetailDeath.writeLock().lock();
        try {
            List<JobDetailShell> jobDetailShellDeathQueue = jobDetailDeathTable.get(jobDetailName);
            if (CollectionUtils.isNotEmpty(jobDetailShellDeathQueue)) {
                for (JobDetailShell jobDetailShell : jobDetailShellDeathQueue) {
                    if (StringUtils.isNoneBlank(id) && jobDetailShell.getId().equals(id)) {
                        boolean result = jobDetailShellDeathQueue.remove(jobDetailShell);
                        removeJobDetailTrack(jobDetailShell);
                        if (result) {
                            LOG.info("JOB-PROCESS-QUEUE: release and remove a death job detail[{}] id=[{}] opaque=[{}]",
                                jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(),
                                jobDetailShell.getRemotingOpaque());
                        } else {
                            LOG.warn(
                                "JOB-PROCESS-QUEUE: release and remove a death job detail[{}] id=[{}] opaque=[{}] fail because isn't found.",
                                jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(),
                                jobDetailShell.getRemotingOpaque());
                        }
                        return jobDetailShell;
                    }
                }
            }
        } finally {
            lockJobDetailDeath.writeLock().unlock();
        }
        return null;
    }
    
    public void removeJobDetailRunning(Channel channel, JobDetailShell jobDetailShell) {
        lockJobDetailRunning.writeLock().lock();
        try {
            LinkedList<JobDetailShell> jobDetailShellRunningList = jobDetailRunningTable.get(channel);
            if (CollectionUtils.isNotEmpty(jobDetailShellRunningList)) {
                boolean result = jobDetailShellRunningList.remove(jobDetailShell);
                if (result) {
                    LOG.debug("JOB-PROCESS-QUEUE: remove a running job detail[{}] id=[{}] opaque=[{}]",
                        jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(), jobDetailShell.getRemotingOpaque());
                }
            }
        } finally {
            lockJobDetailRunning.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param jobDetailShell
     */
    public void completeJobDetailDeath(JobDetailShell jobDetailShell) {
        boolean deathResult = removeJobDetailDeath(jobDetailShell);
        if (deathResult) {
            completeJobDetail(null, jobDetailShell, null);
        }
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param channel
     * @param jobDetailShell
     * @param jee
     */
    /*public void completeJobDetailOld(Channel channel, JobDetailShell jobDetailShell, JobExecutionException jee) {
        lockJobDetailRunning.writeLock().lock();
        try {
            LinkedList<JobDetailShell> jobDetailShellRunningList = jobDetailRunningTable.get(channel);
            if (CollectionUtils.isNotEmpty(jobDetailShellRunningList)) {
                boolean result = jobDetailShellRunningList.remove(jobDetailShell);
                if (result) {
                    completeJobDetail(channel, jobDetailShell, jee);
                    LOG.debug("JOB-PROCESS-QUEUE: complete remove a running job detail[{}] id=[{}] opaque=[{}]",
                        jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(), jobDetailShell.getRemotingOpaque());
                } else {
                    boolean deathResult = removeJobDetailDeath(jobDetailShell);
                    if (deathResult) {
                        completeJobDetail(channel, jobDetailShell, jee);
                    }
                }
            }
        } finally {
            lockJobDetailRunning.writeLock().unlock();
        }
    }*/
    
    /**
     * 
     * 功能说明:
     * 
     * @param channel
     * @param jobDetailName
     * @param id
     * @param jee
     */
    public void completeJobDetail(Channel channel, String jobDetailName, String id, JobExecutionException jee) {
        lockJobDetailRunning.writeLock().lock();
        try {
            LinkedList<JobDetailShell> jobDetailShellRunningList = jobDetailRunningTable.get(channel);
            if (CollectionUtils.isNotEmpty(jobDetailShellRunningList)) {
                boolean result = false;
                Iterator<JobDetailShell> it = jobDetailShellRunningList.iterator();
                while (it.hasNext()) {
                    JobDetailShell jobDetailShell = it.next();
                    if (jobDetailShell.getId().equals(id)) {
                        it.remove();
                        completeJobDetail(channel, jobDetailShell, jee);
                        result = true;
                        LOG.debug("JOB-PROCESS-QUEUE: complete a running job detail[{}] id=[{}] opaque=[{}]",
                            jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(),
                            jobDetailShell.getRemotingOpaque());
                        break;
                    }
                }
                
                if (!result) {
                    JobDetailShell jobDetailShell = removeJobDetailDeath(jobDetailName, id);
                    if (jobDetailShell != null) {
                        completeJobDetail(channel, jobDetailShell, jee);
                    }
                }
            } else {
                // 如果该channel没有在运行中的子任务，那么代表当前子任务之前运行的channel被关闭过，目前的是新生成的channel，需要根据id查找之前的channel
                Iterator<Entry<Channel, LinkedList<JobDetailShell>>> it = jobDetailRunningTable.entrySet().iterator();
                boolean result = false;
                while (it.hasNext()) {
                    Entry<Channel, LinkedList<JobDetailShell>> entry = it.next();
                    Channel rchannel = entry.getKey();
                    LinkedList<JobDetailShell> jobDetailShellList = entry.getValue();
                    
                    if(CollectionUtils.isEmpty(jobDetailShellList)){
                        it.remove();
                        continue;
                    }
                    
                    Iterator<JobDetailShell> jobDetailShellIt = jobDetailShellList.iterator();
                    while(jobDetailShellIt.hasNext()){
                        JobDetailShell jobDetailShell = jobDetailShellIt.next();
                        if (jobDetailShell.getId().equals(id)) {
                            jobDetailShellIt.remove();
                            completeJobDetail(rchannel, jobDetailShell, jee);
                            result = true;
                            LOG.debug("JOB-PROCESS-QUEUE: complete a running job detail[{}] id=[{}] opaque=[{}]",
                                jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(),
                                jobDetailShell.getRemotingOpaque());
                            break;
                        }
                    }
                }
                
                if (!result) {
                    JobDetailShell jobDetailShell = removeJobDetailDeath(jobDetailName, id);
                    if (jobDetailShell != null) {
                        completeJobDetail(channel, jobDetailShell, jee);
                    }
                }
            }
        } finally {
            lockJobDetailRunning.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param channel
     * @param jobDetailShell
     * @param jee
     */
    private void completeJobDetail(Channel channel, JobDetailShell jobDetailShell, JobExecutionException jee) {
        jobDetailShell.updateCompleteTime();
        if (jee != null) {
            if (jee.refireImmediately()) {
                TspJobExecutionContext<Object> tspJobExecutionContext = (TspJobExecutionContext<Object>) jobDetailShell.getContext();
                tspJobExecutionContext.incrementRefireCount();
                updateCompleteStatus(channel, jobDetailShell);
                putJobDetailRetry(jobDetailShell);
            } else {
                jobDetailShell.callBack(jee);
                removeJobDetailTrack(jobDetailShell);
                addCount(jobDetailShell);
            }
        } else {
            LOG.debug("JOB-PROCESS-QUEUE: callback a quartz job detail[{}] id=[{}] opaque=[{}]",
                jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(), jobDetailShell.getRemotingOpaque());
            jobDetailShell.callBack();
            removeJobDetailTrack(jobDetailShell);
            addCount(jobDetailShell);
        }
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param jobDetailShell
     */
    private void addCount(JobDetailShell jobDetailShell) {
        currentSize.decrementAndGet();
        AtomicLong times = runtimesMap.get(jobDetailShell.getJobDetailName());
        if (times == null) {
            synchronized (this) {
                times = runtimesMap.get(jobDetailShell.getJobDetailName());
                if (times == null) {
                    times = new AtomicLong();
                    runtimesMap.put(jobDetailShell.getJobDetailName(), times);
                }
            }
        }
        times.incrementAndGet();
    }
    
    /**
     * 
     * 功能说明:删除子任务轨迹
     * 
     * @param jobDetailShell
     */
    public void removeJobDetailTrack(JobDetailShell jobDetailShell) {
        lockJobDetailTrack.writeLock().lock();
        try {
            if (jobDetailTrackTable.remove(jobDetailShell.getId()) != null) {
                LOG.debug("JOB-PROCESS-QUEUE: remove a job detail track [{}] id=[{}] opaque=[{}].",
                    jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(), jobDetailShell.getRemotingOpaque());
            }
        } finally {
            lockJobDetailTrack.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:获取死任务
     * 
     * @param jobDetailName 取任务的数量
     * @return 子任务运行内核数组
     */
    public List<JobDetailShell> takeJobDetailDeath(String jobDetailName) {
        List<JobDetailShell> result = new ArrayList<JobDetailShell>();
        lockJobDetailDeath.writeLock().lock();
        try {
            LinkedList<JobDetailShell> jobDetailDeathQueue = jobDetailDeathTable.remove(jobDetailName);
            if (CollectionUtils.isNotEmpty(jobDetailDeathQueue)) {
                result.addAll(jobDetailDeathQueue);
            }
        } finally {
            lockJobDetailDeath.writeLock().unlock();
        }
        return result;
    }
    
    /**
     * 
     * 功能说明:获取死任务
     * 
     * @param jobDetailName 取任务的数量
     * @return 子任务运行内核数组
     */
    public List<JobDetailShell> getJobDetailDeath(String jobDetailName) {
        List<JobDetailShell> result = new ArrayList<JobDetailShell>();
        lockJobDetailDeath.writeLock().lock();
        try {
            LinkedList<JobDetailShell> jobDetailDeathQueue = jobDetailDeathTable.get(jobDetailName);
            if (CollectionUtils.isNotEmpty(jobDetailDeathQueue)) {
                result.addAll(jobDetailDeathQueue);
            }
        } finally {
            lockJobDetailDeath.writeLock().unlock();
        }
        return result;
    }
    
    /**
     * 
     * 功能说明:取出所有待执行子任务
     * 
     * @author ZHONGFUHUA
     *
     * @Date 2016年3月8日
     *
     */
    public List<JobDetailShell> takeJobDetail() {
        List<JobDetailShell> result = new ArrayList<JobDetailShell>();
        lockJobDetail.writeLock().lock();
        try {
            result.addAll(jobDetailQueue);
            jobDetailQueue.clear();
        } finally {
            lockJobDetail.writeLock().unlock();
        }
        return result;
    }
    
    /**
     * 
     * 功能说明:出对应数量的待执行子任务
     * 
     * @param batchSize 取任务的数量
     * @return 子任务运行内核数组
     */
    public List<JobDetailShell> takeJobDetail(int batchSize) {
        List<JobDetailShell> result = new ArrayList<JobDetailShell>();
        lockJobDetail.writeLock().lock();
        try {
            if (!jobDetailQueue.isEmpty()) {
                for (int i = 0; i < batchSize; i++) {
                    JobDetailShell jobDetailShell = jobDetailQueue.pollFirst();
                    if (jobDetailShell != null) {
                        result.add(jobDetailShell);
                    } else {
                        break;
                    }
                }
            }
        } finally {
            lockJobDetail.writeLock().unlock();
        }
        return result;
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param channel
     * @param jobDetailShell
     */
    public void updateReciveStatus(Channel channel, JobDetailShell jobDetailShell) {
        lockJobDetailTrack.writeLock().lock();
        try {
            JobDetailTrack jobDetailTrack = jobDetailTrackTable.get(jobDetailShell.getId());
            if (jobDetailTrack != null) {
                jobDetailTrack.updateReciveStatus(channel, jobDetailShell);
            } else {
                LOG.debug("JOB-PROCESS-QUEUE: update recive status fail because track non-existent [{}] id=[{}] opaque=[{}].",
                    jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(), jobDetailShell.getRemotingOpaque());
            }
        } finally {
            lockJobDetailTrack.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param channel
     * @param jobDetailShell
     */
    public void updateCompleteStatus(Channel channel, JobDetailShell jobDetailShell) {
        lockJobDetailTrack.writeLock().lock();
        try {
            JobDetailTrack jobDetailTrack = jobDetailTrackTable.get(jobDetailShell.getId());
            jobDetailTrack.updateCompleteStatus(channel, jobDetailShell);
        } finally {
            lockJobDetailTrack.writeLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:扫描超时执行的子任务
     *
     */
    /*public List<JobDetailShell> getJobDetailOffline() {
        List<JobDetailShell> jobDetailShellOffline = new ArrayList<JobDetailShell>();
        lockJobDetailRunning.writeLock().lock();
        try {
            Iterator<LinkedList<JobDetailShell>> jobDetailRunningTableIt = jobDetailRunningTable.values().iterator();
            while (jobDetailRunningTableIt.hasNext()) {
                LinkedList<JobDetailShell> jobDetailShellList = jobDetailRunningTableIt.next();
                if (CollectionUtils.isNotEmpty(jobDetailShellList)) {
                    for (JobDetailShell jobDetailShell : jobDetailShellList) {
                        if (jobDetailShell.getStatus() == JobDetailRunStatus.OFFLINE
                            || jobDetailShell.getStatus() == JobDetailRunStatus.OFFLINE_TIMEOUT) {
                            jobDetailShellOffline.add(jobDetailShell);
                        }
                    }
                }
            }
        } finally {
            lockJobDetailRunning.writeLock().unlock();
        }
        return jobDetailShellOffline;
    }*/
    
    /**
     * 
     * 功能说明:扫描超时执行的子任务
     *
     */
    public List<JobDetailShell> getJobDetailTimeout() {
        List<JobDetailShell> jobDetailShellTimeout = new ArrayList<JobDetailShell>();
        lockJobDetailRunning.writeLock().lock();
        try {
            Iterator<LinkedList<JobDetailShell>> jobDetailRunningTableIt = jobDetailRunningTable.values().iterator();
            while (jobDetailRunningTableIt.hasNext()) {
                LinkedList<JobDetailShell> jobDetailShellList = jobDetailRunningTableIt.next();
                if (CollectionUtils.isNotEmpty(jobDetailShellList)) {
                    for (JobDetailShell jobDetailShell : jobDetailShellList) {
                        long diff = System.currentTimeMillis() - jobDetailShell.getSendTime().getTime();
                        if (diff > timeout && timeout != -1) {
                            jobDetailShellTimeout.add(jobDetailShell);
                            jobDetailShell.updateTimeout();
                            LOG.warn("JOB-PROCESS-QUEUE: scan a timeout[{}ms] job detail[{}] id=[{}] opaque=[{}]", timeout,
                                jobDetailShell.getContext().getTspJobInstance().getJobKey(), jobDetailShell.getId(),
                                jobDetailShell.getRemotingOpaque());
                        }
                    }
                }
            }
        } finally {
            lockJobDetailRunning.writeLock().unlock();
        }
        return jobDetailShellTimeout;
    }
    
    /**
     * 
     * 功能说明:获取重试子任务的数量大小
     * 
     * @return 重试子任务的数量大小
     */
    public int getJobDetailRetrySpan() {
        lockJobDetailRetry.readLock().lock();
        try {
            return jobDetailRetryQueue.size();
        } finally {
            lockJobDetailRetry.readLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:获取待执行子任务数量大小
     * 
     * @return 待执行子任务数量大小
     */
    public int getJobDetailSpan() {
        lockJobDetail.readLock().lock();
        try {
            return jobDetailQueue.size();
        } finally {
            lockJobDetail.readLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:获取正在运行的子任务数量
     * 
     * @return 正在运行的子任务数量
     */
    public int getJobDetailRunningSpan() {
        int jobDetailRunningTotal = 0;
        lockJobDetailRunning.readLock().lock();
        try {
            Collection<LinkedList<JobDetailShell>> jobDetailRunningList = jobDetailRunningTable.values();
            if (CollectionUtils.isNotEmpty(jobDetailRunningList)) {
                for (LinkedList<JobDetailShell> jobDetailShellList : jobDetailRunningList) {
                    jobDetailRunningTotal += jobDetailShellList.size();
                }
            }
        } finally {
            lockJobDetailRunning.readLock().unlock();
        }
        return jobDetailRunningTotal;
    }
    
    /**
     * 
     * 功能说明:获取执行端正在运行的子任务数量
     * 
     * @param channel 通道
     * @return 正在运行的子任务数量
     */
    public int getJobDetailRunningSpan(Channel channel) {
        lockJobDetailRunning.readLock().lock();
        try {
            LinkedList<JobDetailShell> jobDetailShellList = jobDetailRunningTable.get(channel);
            if (CollectionUtils.isEmpty(jobDetailShellList)) {
                return 0;
            } else {
                return jobDetailShellList.size();
            }
        } finally {
            lockJobDetailRunning.readLock().unlock();
        }
    }
    
    /**
     * 
     * 功能说明:获取正在运行的任务
     * 
     * @return
     */
    public List<JobDetailShell> getJobDetailWaitting() {
        lockJobDetail.readLock().lock();
        List<JobDetailShell> jobDetailWaitingList = new ArrayList<JobDetailShell>();
        try {
            jobDetailWaitingList.addAll(jobDetailQueue);
        } finally {
            lockJobDetail.readLock().unlock();
        }
        return jobDetailWaitingList;
    }
    
    /**
     * 
     * 功能说明:获取正在运行的任务
     * 
     * @return
     */
    public List<JobDetailShell> getJobDetailRetry() {
        lockJobDetailRetry.readLock().lock();
        List<JobDetailShell> jobDetailRetryList = new ArrayList<JobDetailShell>();
        try {
            jobDetailRetryList.addAll(jobDetailRetryQueue);
        } finally {
            lockJobDetailRetry.readLock().unlock();
        }
        return jobDetailRetryList;
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @return
     */
    public List<JobDetailShell> getJobDetailRunning() {
        List<JobDetailShell> jobDetailList = new ArrayList<JobDetailShell>();
        lockJobDetailRunning.readLock().lock();
        try {
            Collection<LinkedList<JobDetailShell>> jobDetailRunningList = jobDetailRunningTable.values();
            for (LinkedList<JobDetailShell> jobDetailShellList : jobDetailRunningList) {
                jobDetailList.addAll(jobDetailShellList);
            }
        } finally {
            lockJobDetailRunning.readLock().unlock();
        }
        return jobDetailList;
    }
    
    /**
     * 
     * 功能说明:获取正在运行的任务
     * 
     * @return
     */
    public Map<Channel, LinkedList<JobDetailShell>> getJobDetailRunningChannel() {
        lockJobDetailRunning.readLock().lock();
        Map<Channel, LinkedList<JobDetailShell>> jobDetailRunningMap = new HashMap<Channel, LinkedList<JobDetailShell>>();
        try {
            jobDetailRunningMap.putAll(jobDetailRunningTable);
        } finally {
            lockJobDetailRunning.readLock().unlock();
        }
        return jobDetailRunningMap;
    }
    
    /**
     * 功能说明:
     * 
     * @param id
     * @return
     */
    public JobDetailShellChannelPairInfo getJobDetailRunningPairInfo(String id) {
        lockJobDetailRunning.readLock().lock();
        JobDetailShellChannelPairInfo jobDetailShellChannelPairInfo = null;
        try {
            Iterator<Entry<Channel, LinkedList<JobDetailShell>>> it = jobDetailRunningTable.entrySet().iterator();
            while (it.hasNext()) {
                Entry<Channel, LinkedList<JobDetailShell>> entry = it.next();
                Channel channel = entry.getKey();
                LinkedList<JobDetailShell> jobDetailShellList = entry.getValue();
                for (JobDetailShell jobDetailShell : jobDetailShellList) {
                    if (jobDetailShell.getId().equals(id)) {
                        jobDetailShellChannelPairInfo = new JobDetailShellChannelPairInfo(channel, jobDetailShell);
                        break;
                    }
                }
            }
        } finally {
            lockJobDetailRunning.readLock().unlock();
        }
        return jobDetailShellChannelPairInfo;
    }
    
    /**
     * 
     * 功能说明:获取子任务已经执行过的Channel
     * 
     * @param jobDetailShell
     * @return 已经执行过的Channel
     */
    public JobDetailTrack getJobDetailTrack(String id) {
        lockJobDetailTrack.readLock().lock();
        JobDetailTrack jobDetailTrack = null;
        try {
            jobDetailTrack = jobDetailTrackTable.get(id);
        } finally {
            lockJobDetailTrack.readLock().unlock();
        }
        return jobDetailTrack;
    }
    
    /**
     * 
     * 功能说明:获取子任务已经执行过的Channel
     * 
     * @param jobDetailShell
     * @return 已经执行过的Channel
     */
    public List<Channel> getChannelsRunBefore(JobDetailShell jobDetailShell) {
        List<Channel> channelList = new ArrayList<Channel>();
        lockJobDetailTrack.readLock().lock();
        try {
            JobDetailTrack jobDetailTrack = jobDetailTrackTable.get(jobDetailShell.getId());
            if (jobDetailTrack != null) {
                channelList.addAll(jobDetailTrack.getChannelInfo().keySet());
            }
        } finally {
            lockJobDetailTrack.readLock().unlock();
        }
        return channelList;
    }
    
    /**
     * 
     * 功能说明:获取子任务最后执行过的Channel
     * 
     * @param jobDetailShell
     * @return 已经执行过的Channel
     */
    public Channel getJobDetailRunningLastChannel(JobDetailShell jobDetailShell) {
        lockJobDetailTrack.readLock().lock();
        try {
            JobDetailTrack jobDetailTrack = jobDetailTrackTable.get(jobDetailShell.getId());
            if (jobDetailTrack != null) {
                return jobDetailTrack.getLastchannel();
            }
        } finally {
            lockJobDetailTrack.readLock().unlock();
        }
        return null;
    }
    
    /**
     * 
     * 功能说明:获取子任务最后执行过的Channel
     * 
     * @param jobDetailShell
     * @return 已经执行过的Channel
     */
    public Channel getJobDetailRunningLastSecondChannel(JobDetailShell jobDetailShell) {
        lockJobDetailTrack.readLock().lock();
        try {
            JobDetailTrack jobDetailTrack = jobDetailTrackTable.get(jobDetailShell.getId());
            if (jobDetailTrack != null) {
                return jobDetailTrack.getLastSecondchannel();
            }
        } finally {
            lockJobDetailTrack.readLock().unlock();
        }
        return null;
    }
    
    
    
    /**
     * 
     * 功能说明:获取子任务已经执行过的Channel
     * 
     * @param jobDetailShell
     * @return 已经执行过的Channel
     */
    public Date getJobDetailDeathLastDate(JobDetailShell jobDetailShell) {
        lockJobDetailTrack.readLock().lock();
        try {
            JobDetailTrack jobDetailTrack = jobDetailTrackTable.get(jobDetailShell.getId());
            if (jobDetailTrack != null) {
                return jobDetailTrack.getLastDate();
            }
        } finally {
            lockJobDetailTrack.readLock().unlock();
        }
        return null;
    }
    
    /**
     * 
     * 功能说明:获取通道对应运行的子任务数量
     * 
     * @param channelList 任务可发送的通道数组
     * @return 通道对应运行的子任务数量Map
     */
    public Map<Channel, Integer> getChannelsRunningNum(List<Channel> channelList) {
        Map<Channel, Integer> channelRefJobDetialNumMap = new HashMap<Channel, Integer>();
        lockJobDetailRunning.readLock().lock();
        try {
            if (CollectionUtils.isNotEmpty(channelList)) {
                for (Channel channel : channelList) {
                    LinkedList<JobDetailShell> jobDetailShellList = jobDetailRunningTable.get(channel);
                    if (CollectionUtils.isNotEmpty(jobDetailShellList)) {
                        channelRefJobDetialNumMap.put(channel, jobDetailShellList.size());
                    } else {
                        channelRefJobDetialNumMap.put(channel, 0);
                    }
                }
            }
        } finally {
            lockJobDetailRunning.readLock().unlock();
        }
        return channelRefJobDetialNumMap;
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @return
     */
    public List<String> getJobDetailDeathNames() {
        List<String> jobDetailNameList = new ArrayList<String>();
        lockJobDetailDeath.readLock().lock();
        try {
            jobDetailNameList.addAll(this.jobDetailDeathTable.keySet());
        } finally {
            lockJobDetailDeath.readLock().unlock();
        }
        return jobDetailNameList;
    }
    
    /**
     * 
     * 功能说明:设置该任务的超时时间
     * 
     * @param jobTimeout 超时时间
     */
    public void updateTimeout(long jobTimeout) {
        if (jobTimeout > 0 || jobTimeout == -1) {
            this.timeout = jobTimeout;
        }
    }
    
    /**
     * 
     * 功能说明:设置任务的超时策略
     * 
     * @param timeoutPolicy 超时策略
     */
    public void updateTimeoutPolicy(short timeoutPolicy) {
        this.timeoutPolicy = timeoutPolicy;
    }
    
    /**
     * 功能说明: 设置死任务重置延迟时间-1:不延迟
     *
     * @param resetDeathDelayTime 死任务重置延迟时间-1:不延迟
     */
    public void updateResetDeathDelayTime(long resetDeathDelayTime) {
        this.resetDeathDelayTime = resetDeathDelayTime;
    }
    
    /**
     * 功能说明: 设置任务数量缓冲区大小
     *
     * @param bufferLimitSize 任务数量缓冲区大小 
     */
    public void updateBufferLimitSize(int bufferLimitSize) {
        this.bufferLimitSize = bufferLimitSize;
    }
    
    /**
     * 功能说明: 获取死任务重置延迟时间-1:不延迟
     *
     * @return resetDeathDelayTime 死任务重置延迟时间-1:不延迟
     */
    public long getResetDeathDelayTime() {
        return resetDeathDelayTime;
    }
    
    /**
     * 
     * 功能说明:设置任务的超时策略
     * 
     * @param timeoutPolicy 超时策略
     * @return 
     */
    public short getTimeoutPolicy() {
        return this.timeoutPolicy;
    }
    
    /**
     * 功能说明: 获取任务名
     *
     * @return jobName 任务名
     */
    public String getJobName() {
        return jobName;
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param name
     * @return
     */
    public long getJobDetailRuntimes(String name) {
        if (this.runtimesMap.get(name) == null) {
            return 0;
        } else {
            return this.runtimesMap.get(name).longValue();
        }
    }
    
}
